Implements including attachments in notifications (on supported adapters/backends)#3
Implements including attachments in notifications (on supported adapters/backends)#3
Conversation
Reviewer's GuideThis PR adds end-to-end support for attachments in notifications by extending data models, backends, service layer, and adapters, and includes a comprehensive test suite. Attachments are represented by new dataclasses and handled via a FileAttachment interface; FakeFileBackend now stores attachments in memory with checksum and metadata; NotificationService validates and passes attachments to backends; fake adapters capture attachment info; requests dependency is added for URL downloads. Sequence diagram for notification creation with attachmentssequenceDiagram
participant U as actor User
participant S as NotificationService
participant B as NotificationBackend
participant A as NotificationAdapter
U->>S: create_notification(..., attachments)
S->>S: _validate_attachments(attachments)
S->>B: persist_notification(..., attachments)
B->>B: _store_attachments(attachments)
B->>B: store Notification with StoredAttachment(s)
S->>A: send(notification, context)
A->>A: process notification.attachments
Class diagram for notification attachment supportclassDiagram
class NotificationAttachment {
+file: FileAttachment
+filename: str
+content_type: str | None
+description: str | None
+is_inline: bool
+is_url() bool
+_detect_content_type() str
}
class StoredAttachment {
+id: str | uuid.UUID
+filename: str
+content_type: str
+size: int
+checksum: str
+created_at: datetime.datetime
+file: AttachmentFile
+description: str | None
+is_inline: bool
+storage_metadata: dict[str, Any]
+get_file_data() bytes
+get_file_stream() BinaryIO
+get_file_url(expires_in: int) str
+delete() None
}
class AttachmentFile {
<<interface>>
+read() bytes
+stream() BinaryIO
+url(expires_in: int) str
+delete() None
}
class FakeFileAttachmentFile {
+file_data: bytes
+filename: str
+_deleted: bool
+read() bytes
+stream() BinaryIO
+url(expires_in: int) str
+delete() None
}
NotificationAttachment --> AttachmentFile
StoredAttachment --> AttachmentFile
FakeFileAttachmentFile --|> AttachmentFile
class Notification {
+attachments: list[StoredAttachment]
}
class OneOffNotification {
+attachments: list[StoredAttachment]
}
Notification --> StoredAttachment
OneOffNotification --> StoredAttachment
Class diagram for FakeEmailAdapter and sent_emails structureclassDiagram
class FakeEmailAdapter {
+sent_emails: list[tuple[Notification | OneOffNotification, NotificationContextDict, list[dict]]]
+send(notification, context) None
}
class Notification {
+attachments: list[StoredAttachment]
}
FakeEmailAdapter --> Notification
Notification --> StoredAttachment
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #3 +/- ##
==========================================
- Coverage 84.17% 83.61% -0.57%
==========================================
Files 22 22
Lines 1125 1294 +169
Branches 75 113 +38
==========================================
+ Hits 947 1082 +135
- Misses 143 167 +24
- Partials 35 45 +10
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
There was a problem hiding this comment.
Hey there - I've reviewed your changes - here's some feedback:
- Extract the shared file-reading and URL-handling logic in _read_attachment_data, _is_url, and _download_from_url into a common utility to eliminate duplication across backends and the service layer.
- Consolidate the duplicate _store_attachments implementations in FakeFileBackend and FakeAsyncIOFileBackend into a single base method to adhere to DRY principles.
- In the async service layer, the synchronous requests.get call in _download_from_url can block the event loop—consider using an async HTTP client or offloading the download to a thread pool.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- Extract the shared file-reading and URL-handling logic in _read_attachment_data, _is_url, and _download_from_url into a common utility to eliminate duplication across backends and the service layer.
- Consolidate the duplicate _store_attachments implementations in FakeFileBackend and FakeAsyncIOFileBackend into a single base method to adhere to DRY principles.
- In the async service layer, the synchronous requests.get call in _download_from_url can block the event loop—consider using an async HTTP client or offloading the download to a thread pool.
## Individual Comments
### Comment 1
<location> `vintasend/tests/test_services/test_notification_attachments.py:217-220` </location>
<code_context>
+ retrieved_data = stored_attachment.get_file_data()
+ assert retrieved_data == test_data
+
+ def test_store_attachments_with_url(self):
+ """Test storing attachments from URLs (using fake download)"""
+ url = "http://example.com/document.pdf"
+ attachment = NotificationAttachment(
+ filename="document.pdf",
+ file=url,
+ )
+
+ stored = self.backend._store_attachments([attachment])
+
+ assert len(stored) == 1
+ stored_attachment = stored[0]
+ assert stored_attachment.filename == "document.pdf"
+
+ # Should contain fake downloaded content
+ retrieved_data = stored_attachment.get_file_data()
+ assert b"Downloaded content from" in retrieved_data
+
+ def test_store_attachments_empty_list(self):
</code_context>
<issue_to_address>
**suggestion (testing):** Consider adding a test for unsupported URL schemes.
Adding a test for an unsupported URL scheme, such as 'ftp://', will help verify that the backend correctly raises a ValueError or handles the input as expected.
```suggestion
def test_store_attachments_unsupported_url_scheme(self):
"""Test storing attachments with unsupported URL scheme should raise ValueError"""
url = "ftp://example.com/document.pdf"
attachment = NotificationAttachment(
filename="document.pdf",
file=url,
)
with pytest.raises(ValueError):
self.backend._store_attachments([attachment])
def test_store_attachments_empty_list(self):
"""Test storing empty attachment list"""
stored = self.backend._store_attachments([])
assert stored == []
```
</issue_to_address>
### Comment 2
<location> `vintasend/tests/test_services/test_notification_attachments.py:701-710` </location>
<code_context>
+ def test_notification_attachment_file_types(self):
</code_context>
<issue_to_address>
**suggestion (testing):** Missing test for Path object as file input.
Please add a test case using pathlib.Path to verify support for Path objects as file inputs.
</issue_to_address>
### Comment 3
<location> `vintasend/tests/test_services/test_notification_attachments.py:721-730` </location>
<code_context>
+ def test_content_type_detection_accuracy(self):
</code_context>
<issue_to_address>
**suggestion (testing):** Consider adding a test for files with no extension.
Please add a test for filenames without an extension to confirm that 'application/octet-stream' is used as the default content type.
</issue_to_address>
### Comment 4
<location> `vintasend/tests/test_services/test_notification_attachments.py:179-177` </location>
<code_context>
+ retrieved_data = stored_attachment.get_file_data()
+ assert retrieved_data == test_data
+
+ def test_store_attachments_with_file_like_object(self):
+ """Test storing attachments with file-like objects"""
+ test_data = b"File-like object content"
+ file_obj = io.BytesIO(test_data)
+
+ attachment = NotificationAttachment(
+ filename="filelike.txt",
+ file=file_obj,
+ )
+
+ stored = self.backend._store_attachments([attachment])
+
+ assert len(stored) == 1
+ stored_attachment = stored[0]
+ assert stored_attachment.size == len(test_data)
+
+ # Verify file data
+ retrieved_data = stored_attachment.get_file_data()
+ assert retrieved_data == test_data
+
+ def test_store_attachments_with_url(self):
</code_context>
<issue_to_address>
**suggestion (testing):** Consider adding a test for io.StringIO as file input.
Please add a test that uses io.StringIO to ensure attachments from StringIO objects are processed and encoded to bytes as expected.
</issue_to_address>
### Comment 5
<location> `vintasend/tests/test_services/test_notification_attachments.py:97-102` </location>
<code_context>
</code_context>
<issue_to_address>
**issue (code-quality):** Avoid loops in tests. ([`no-loop-in-tests`](https://docs.sourcery.ai/Reference/Rules-and-In-Line-Suggestions/Python/Default-Rules/no-loop-in-tests))
<details><summary>Explanation</summary>Avoid complex code, like loops, in test functions.
Google's software engineering guidelines says:
"Clear tests are trivially correct upon inspection"
To reach that avoid complex code in tests:
* loops
* conditionals
Some ways to fix this:
* Use parametrized tests to get rid of the loop.
* Move the complex logic into helpers.
* Move the complex part into pytest fixtures.
> Complexity is most often introduced in the form of logic. Logic is defined via the imperative parts of programming languages such as operators, loops, and conditionals. When a piece of code contains logic, you need to do a bit of mental computation to determine its result instead of just reading it off of the screen. It doesn't take much logic to make a test more difficult to reason about.
Software Engineering at Google / [Don't Put Logic in Tests](https://abseil.io/resources/swe-book/html/ch12.html#donapostrophet_put_logic_in_tests)
</details>
</issue_to_address>
### Comment 6
<location> `vintasend/tests/test_services/test_notification_attachments.py:712-719` </location>
<code_context>
</code_context>
<issue_to_address>
**issue (code-quality):** Avoid loops in tests. ([`no-loop-in-tests`](https://docs.sourcery.ai/Reference/Rules-and-In-Line-Suggestions/Python/Default-Rules/no-loop-in-tests))
<details><summary>Explanation</summary>Avoid complex code, like loops, in test functions.
Google's software engineering guidelines says:
"Clear tests are trivially correct upon inspection"
To reach that avoid complex code in tests:
* loops
* conditionals
Some ways to fix this:
* Use parametrized tests to get rid of the loop.
* Move the complex logic into helpers.
* Move the complex part into pytest fixtures.
> Complexity is most often introduced in the form of logic. Logic is defined via the imperative parts of programming languages such as operators, loops, and conditionals. When a piece of code contains logic, you need to do a bit of mental computation to determine its result instead of just reading it off of the screen. It doesn't take much logic to make a test more difficult to reason about.
Software Engineering at Google / [Don't Put Logic in Tests](https://abseil.io/resources/swe-book/html/ch12.html#donapostrophet_put_logic_in_tests)
</details>
</issue_to_address>
### Comment 7
<location> `vintasend/tests/test_services/test_notification_attachments.py:736-742` </location>
<code_context>
</code_context>
<issue_to_address>
**issue (code-quality):** Avoid loops in tests. ([`no-loop-in-tests`](https://docs.sourcery.ai/Reference/Rules-and-In-Line-Suggestions/Python/Default-Rules/no-loop-in-tests))
<details><summary>Explanation</summary>Avoid complex code, like loops, in test functions.
Google's software engineering guidelines says:
"Clear tests are trivially correct upon inspection"
To reach that avoid complex code in tests:
* loops
* conditionals
Some ways to fix this:
* Use parametrized tests to get rid of the loop.
* Move the complex logic into helpers.
* Move the complex part into pytest fixtures.
> Complexity is most often introduced in the form of logic. Logic is defined via the imperative parts of programming languages such as operators, loops, and conditionals. When a piece of code contains logic, you need to do a bit of mental computation to determine its result instead of just reading it off of the screen. It doesn't take much logic to make a test more difficult to reason about.
Software Engineering at Google / [Don't Put Logic in Tests](https://abseil.io/resources/swe-book/html/ch12.html#donapostrophet_put_logic_in_tests)
</details>
</issue_to_address>
### Comment 8
<location> `vintasend/services/notification_adapters/stubs/fake_adapter.py:38` </location>
<code_context>
def send(self, notification: "Notification | OneOffNotification", context: "NotificationContextDict") -> None:
self.template_renderer.render(notification, context)
# Capture attachment information for testing
attachment_info = []
for attachment in notification.attachments:
attachment_info.append({
'id': str(attachment.id),
'filename': attachment.filename,
'content_type': attachment.content_type,
'size': attachment.size,
'is_inline': attachment.is_inline,
'description': attachment.description,
'checksum': attachment.checksum,
})
self.sent_emails.append((notification, context, attachment_info))
</code_context>
<issue_to_address>
**issue (code-quality):** Convert for loop into list comprehension ([`list-comprehension`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/list-comprehension/))
</issue_to_address>
### Comment 9
<location> `vintasend/services/notification_adapters/stubs/fake_adapter.py:69` </location>
<code_context>
async def send(self, notification: "Notification | OneOffNotification", context: "NotificationContextDict") -> None:
self.template_renderer.render(notification, context)
# Capture attachment information for testing
attachment_info = []
for attachment in notification.attachments:
attachment_info.append({
'id': str(attachment.id),
'filename': attachment.filename,
'content_type': attachment.content_type,
'size': attachment.size,
'is_inline': attachment.is_inline,
'description': attachment.description,
'checksum': attachment.checksum,
})
self.sent_emails.append((notification, context, attachment_info))
</code_context>
<issue_to_address>
**issue (code-quality):** Convert for loop into list comprehension ([`list-comprehension`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/list-comprehension/))
</issue_to_address>
### Comment 10
<location> `vintasend/services/notification_backends/stubs/fake_backend.py:301` </location>
<code_context>
def _read_attachment_data(self, file) -> bytes:
"""Read file data from various file-like object types"""
from pathlib import Path
if isinstance(file, bytes):
return file
elif isinstance(file, str):
if self._is_url(file):
return self._download_from_url(file)
else:
# Read from file path
with open(file, 'rb') as f:
return f.read()
elif isinstance(file, Path):
with open(file, 'rb') as f:
return f.read()
elif hasattr(file, 'read'):
current_pos = file.tell() if hasattr(file, 'tell') else 0
if hasattr(file, 'seek'):
file.seek(0)
data = file.read()
if hasattr(file, 'seek'):
file.seek(current_pos)
if isinstance(data, str):
return data.encode('utf-8')
return data
else:
raise ValueError(f"Unsupported file type: {type(file)}")
</code_context>
<issue_to_address>
**issue (code-quality):** We've found these issues:
- Remove unnecessary else after guard condition ([`remove-unnecessary-else`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/remove-unnecessary-else/))
- Extract code out into method ([`extract-method`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/extract-method/))
- Lift code into else after jump in control flow ([`reintroduce-else`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/reintroduce-else/))
- Replace if statement with if expression ([`assign-if-exp`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/assign-if-exp/))
</issue_to_address>
### Comment 11
<location> `vintasend/services/notification_backends/stubs/fake_backend.py:548` </location>
<code_context>
def _read_attachment_data(self, file) -> bytes:
"""Read file data from various file-like object types"""
from pathlib import Path
if isinstance(file, bytes):
return file
elif isinstance(file, str):
if self._is_url(file):
return self._download_from_url(file)
else:
# Read from file path
with open(file, 'rb') as f:
return f.read()
elif isinstance(file, Path):
with open(file, 'rb') as f:
return f.read()
elif hasattr(file, 'read'):
current_pos = file.tell() if hasattr(file, 'tell') else 0
if hasattr(file, 'seek'):
file.seek(0)
data = file.read()
if hasattr(file, 'seek'):
file.seek(current_pos)
if isinstance(data, str):
return data.encode('utf-8')
return data
else:
raise ValueError(f"Unsupported file type: {type(file)}")
</code_context>
<issue_to_address>
**issue (code-quality):** We've found these issues:
- Remove unnecessary else after guard condition ([`remove-unnecessary-else`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/remove-unnecessary-else/))
- Extract code out into method ([`extract-method`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/extract-method/))
- Lift code into else after jump in control flow ([`reintroduce-else`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/reintroduce-else/))
- Replace if statement with if expression ([`assign-if-exp`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/assign-if-exp/))
</issue_to_address>
### Comment 12
<location> `vintasend/services/notification_service.py:185` </location>
<code_context>
def _read_file_data(self, file) -> bytes:
"""Read file data from various file-like object types"""
from pathlib import Path
if isinstance(file, str):
if self._is_url(file):
return self._download_from_url(file)
else:
# Read from file path
with open(file, 'rb') as f:
return f.read()
elif isinstance(file, Path):
with open(file, 'rb') as f:
return f.read()
elif hasattr(file, 'read'):
current_pos = file.tell() if hasattr(file, 'tell') else 0
if hasattr(file, 'seek'):
file.seek(0)
data = file.read()
if hasattr(file, 'seek'):
file.seek(current_pos)
if isinstance(data, str):
return data.encode('utf-8')
return data
else:
raise ValueError(f"Unsupported file type: {type(file)}")
</code_context>
<issue_to_address>
**issue (code-quality):** We've found these issues:
- Merge duplicate blocks in conditional ([`merge-duplicate-blocks`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/merge-duplicate-blocks/))
- Extract code out into method ([`extract-method`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/extract-method/))
- Lift code into else after jump in control flow ([`reintroduce-else`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/reintroduce-else/))
- Replace if statement with if expression ([`assign-if-exp`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/assign-if-exp/))
- Remove redundant conditional ([`remove-redundant-if`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/remove-redundant-if/))
</issue_to_address>
### Comment 13
<location> `vintasend/services/notification_service.py:741` </location>
<code_context>
def _read_file_data(self, file) -> bytes:
"""Read file data from various file-like object types"""
from pathlib import Path
if isinstance(file, str):
if self._is_url(file):
return self._download_from_url(file)
else:
# Read from file path
with open(file, 'rb') as f:
return f.read()
elif isinstance(file, Path):
with open(file, 'rb') as f:
return f.read()
elif hasattr(file, 'read'):
current_pos = file.tell() if hasattr(file, 'tell') else 0
if hasattr(file, 'seek'):
file.seek(0)
data = file.read()
if hasattr(file, 'seek'):
file.seek(current_pos)
if isinstance(data, str):
return data.encode('utf-8')
return data
else:
raise ValueError(f"Unsupported file type: {type(file)}")
</code_context>
<issue_to_address>
**issue (code-quality):** We've found these issues:
- Merge duplicate blocks in conditional ([`merge-duplicate-blocks`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/merge-duplicate-blocks/))
- Extract code out into method ([`extract-method`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/extract-method/))
- Lift code into else after jump in control flow ([`reintroduce-else`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/reintroduce-else/))
- Replace if statement with if expression ([`assign-if-exp`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/assign-if-exp/))
- Remove redundant conditional ([`remove-redundant-if`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/remove-redundant-if/))
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Summary by Sourcery
Implement attachments support throughout the notification system: update data models to include attachments, enhance backends to store and retrieve attachments, extend service and adapter layers to accept and process attachments, and add comprehensive tests
New Features:
Enhancements:
Build:
Tests: